package com.xy.webadmin.app.config;

import com.xy.webadmin.app.constant.MqttConstants;
import com.xy.webadmin.app.model.MqttBrokerConfig;
import com.xy.webadmin.app.service.MqttBrokerConfigService;
import com.xy.webadmin.app.service.MqttClientManager;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * MQTT订阅配置类 - 基于Spring Integration
 * 用于自动订阅MQTT主题并处理接收到的消息
 *
 * @author xiaoyue
 * @date 2025-07-03
 */
@Slf4j
@Configuration
public class MqttSubscribeAdapterConfig {

    @Autowired
    private MqttClientManager mqttClientManager;
    
    @Autowired
    private MqttBrokerConfigService mqttBrokerConfigService;
    
    @Autowired
    private DefaultPahoMessageConverter messageConverter;
    
    @Autowired
    private MqttPahoClientFactory mqttClientFactory;
    
    @Autowired
    @Qualifier(MqttConstants.SpringIntegration.MQTT_INPUT_CHANNEL)
    private MessageChannel mqttInputChannel;
    
    /**
     * 存储订阅适配器，按配置ID索引
     */
    private final Map<Long, MqttPahoMessageDrivenChannelAdapter> inboundAdapterMap = new ConcurrentHashMap<>();
    
    /**
     * 创建并注册MQTT订阅适配器
     *
     * @param config MQTT配置
     * @param topics 订阅主题列表
     * @param qos QoS等级列表
     * @return 订阅适配器
     */
    public MqttPahoMessageDrivenChannelAdapter createSubscribeAdapter(
            MqttBrokerConfig config, String[] topics, int[] qos) {
        // 检查参数有效性
        if (config == null || config.getId() == null) {
            log.error("MQTT配置为null或ID为null，无法创建订阅适配器");
            return null;
        }
        
        // 检查broker URL是否存在
        if (config.getBrokerUrl() == null || config.getBrokerUrl().isEmpty()) {
            log.error("MQTT代理URL为空，无法创建订阅适配器, configId: {}", config.getId());
            return null;
        }
        
        if (topics == null || topics.length == 0 || qos == null || qos.length == 0) {
            log.error("MQTT主题或QoS数组为空，无法创建订阅适配器, configId: {}", config.getId());
            return null;
        }
        
        if (topics.length != qos.length) {
            log.error("MQTT主题数量与QoS数量不匹配，无法创建订阅适配器, configId: {}", config.getId());
            return null;
        }
        
        // 如果已存在，先停止并移除
        if (inboundAdapterMap.containsKey(config.getId())) {
            MqttPahoMessageDrivenChannelAdapter existingAdapter = inboundAdapterMap.get(config.getId());
            try {
                if (existingAdapter != null) {
                    log.info("停止现有MQTT适配器, configId: {}", config.getId());
                    existingAdapter.stop();
                }
            } catch (Exception e) {
                log.warn("停止现有MQTT适配器时发生异常", e);
            }
            inboundAdapterMap.remove(config.getId());
        }
        
        try {
            // 获取客户端工厂，如果特定配置的工厂不存在，使用默认工厂
            MqttPahoClientFactory clientFactory = mqttClientManager.getMqttClientFactory(config.getId());
            if (clientFactory == null) {
                log.warn("未找到配置ID为{}的MQTT客户端工厂，使用默认工厂", config.getId());
                clientFactory = this.mqttClientFactory;
            }
            
            // 确保客户端ID不为空
            String baseClientId = config.getClientId();
            if (baseClientId == null || baseClientId.isEmpty()) {
                baseClientId = MqttConstants.ConnectionDefaults.CLIENT_ID_PREFIX + config.getId();
                log.warn("MQTT客户端ID为空，使用生成的ID前缀: {}", baseClientId);
            }
            
            // 使用唯一的客户端ID，避免冲突
            String clientId = baseClientId + "_sub_" + System.currentTimeMillis();
            log.info("创建MQTT订阅适配器，使用客户端ID: {}, brokerUrl: {}", clientId, config.getBrokerUrl());
            
            // 创建新的适配器
            MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                    clientId,
                    clientFactory,
                    topics);
            adapter.setQos(qos);
            adapter.setConverter(messageConverter);
            adapter.setOutputChannel(mqttInputChannel);
            adapter.setCompletionTimeout(MqttConstants.ApiEndpoints.HTTP_TIMEOUT);
            
            // 配置适配器
            adapter.setBeanName(MqttConstants.SpringIntegration.BEAN_NAME_INBOUND_ADAPTER + config.getId());
            
            try {
                // 初始化适配器
                log.debug("开始初始化MQTT订阅适配器, configId: {}", config.getId());
                adapter.afterPropertiesSet();
                
                // 尝试启动适配器
                log.info("开始启动MQTT订阅适配器, configId: {}, topics: {}", 
                        config.getId(), String.join(",", topics));
                adapter.start();
                
                // 等待一小段时间确保连接建立
                TimeUnit.SECONDS.sleep(1);
                
                // 存储适配器
                inboundAdapterMap.put(config.getId(), adapter);
                
                log.info("成功创建并启动MQTT订阅适配器, configId: {}, topics: {}", 
                        config.getId(), String.join(",", topics));
                return adapter;
            } catch (Exception e) {
                log.error("MQTT订阅适配器初始化失败, configId: {}, brokerUrl: {}, error: {}", 
                        config.getId(), config.getBrokerUrl(), e.getMessage(), e);
                // 尝试优雅关闭
                try {
                    if (adapter != null) {
                        log.debug("尝试关闭失败的MQTT适配器, configId: {}", config.getId());
                        adapter.destroy();
                    }
                } catch (Exception ex) {
                    log.warn("关闭失败的MQTT适配器时发生异常", ex);
                }
                return null;
            }
        } catch (Exception e) {
            log.error("创建MQTT订阅适配器时发生异常, configId: {}, error: {}", config.getId(), e.getMessage(), e);
            return null;
        }
    }
    
    /**
     * 处理接收到的MQTT消息
     *
     * @return 消息处理器
     */
    @Bean
    @ServiceActivator(inputChannel = MqttConstants.SpringIntegration.MQTT_INPUT_CHANNEL)
    public MessageHandler mqttMessageHandler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                try {
                    if (message == null || message.getHeaders() == null) {
                        log.warn("收到空的MQTT消息或消息头");
                        return;
                    }
                    
                    String topic = (String) message.getHeaders().get(MqttConstants.SpringIntegration.HEADER_RECEIVED_TOPIC);
                    Object payloadObj = message.getPayload();
                    String payload = payloadObj != null ? payloadObj.toString() : "";
                    
                    log.debug("收到MQTT消息: topic={}, payload={}", topic, payload);
                    
                    // 解析主题格式：company/{companyId}/device/{deviceId}/...
                    if (topic != null && topic.startsWith("company/")) {
                        processCompanyMessage(topic, payload);
                    }
                    // 系统主题
                    else if (topic != null && topic.startsWith("$SYS/")) {
                        processSystemMessage(topic, payload);
                    }
                    // 其他主题
                    else {
                        log.info("收到其他主题消息: {}, payload: {}", topic, payload);
                    }
                    
                } catch (Exception e) {
                    log.error("处理MQTT消息失败", e);
                }
            }
        };
    }
    
    /**
     * 处理公司相关的消息
     * 
     * @param topic 主题
     * @param payload 消息内容
     */
    private void processCompanyMessage(String topic, String payload) {
        try {
            String[] parts = topic.split("/");
            if (parts.length >= 2) {
                Long companyId = Long.parseLong(parts[1]);
                
                // 设备状态消息
                if (topic.contains("/status")) {
                    log.info("收到设备状态消息, 公司ID: {}, topic: {}", companyId, topic);
                    // TODO: 处理设备状态消息
                }
                // 设备数据消息
                else if (topic.contains("/data")) {
                    log.info("收到设备数据消息, 公司ID: {}, topic: {}", companyId, topic);
                    // TODO: 处理设备数据消息
                }
                // 命令响应消息
                else if (topic.contains("/response")) {
                    log.info("收到命令响应消息, 公司ID: {}, topic: {}", companyId, topic);
                    // TODO: 处理命令响应消息
                }
                // 其他公司消息
                else {
                    log.info("收到其他公司消息, 公司ID: {}, topic: {}", companyId, topic);
                }
            }
        } catch (Exception e) {
            log.error("处理公司消息失败, topic: {}", topic, e);
        }
    }
    
    /**
     * 处理系统主题消息
     * 
     * @param topic 主题
     * @param payload 消息内容
     */
    private void processSystemMessage(String topic, String payload) {
        // 处理系统主题消息，如$SYS/broker/clients/connected等
        log.info("收到系统主题消息: {}, payload: {}", topic, payload);
    }
    
    /**
     * 注册默认的订阅
     * 可以在应用启动时调用此方法订阅指定主题
     */
    public void registerDefaultSubscriptions() {
        try {
            List<MqttBrokerConfig> configs = mqttBrokerConfigService.getMqttBrokerConfigList(null, null);
            if (configs == null || configs.isEmpty()) {
                log.warn("没有找到MQTT配置，无法注册默认订阅");
                return;
            }
            
            log.info("开始注册默认订阅，共找到{}个MQTT配置", configs.size());
            int successCount = 0;
            
            for (MqttBrokerConfig config : configs) {
                try {
                    if (config == null || config.getId() == null || config.getCompanyId() == null) {
                        log.warn("MQTT配置无效，跳过注册订阅");
                        continue;
                    }
                    
                    // 检查broker URL是否存在
                    if (config.getBrokerUrl() == null || config.getBrokerUrl().isEmpty()) {
                        log.error("MQTT代理URL为空，跳过注册订阅, configId: {}", config.getId());
                        continue;
                    }
                    
                    log.info("开始为配置ID: {}注册默认订阅, brokerUrl: {}", config.getId(), config.getBrokerUrl());
                    
                    // 订阅公司相关主题
                    String companyTopic = "company/" + config.getCompanyId() + "/#";
                    
                    // 订阅系统主题（如果需要）
                    String systemTopic = "$SYS/#";
                    
                    log.info("准备订阅主题: [{}] 和 [{}], configId: {}", companyTopic, systemTopic, config.getId());
                    
                    MqttPahoMessageDrivenChannelAdapter adapter = createSubscribeAdapter(
                            config, 
                            new String[]{companyTopic, systemTopic}, 
                            new int[]{MqttConstants.QosLevel.AT_LEAST_ONCE, MqttConstants.QosLevel.AT_MOST_ONCE} // 公司主题QoS=1，系统主题QoS=0
                    );
                    
                    if (adapter != null) {
                        successCount++;
                        log.info("成功为配置ID: {}注册默认订阅", config.getId());
                    } else {
                        log.warn("为配置ID: {}注册默认订阅失败，adapter创建返回null", config.getId());
                    }
                } catch (Exception e) {
                    log.error("为配置ID: {}注册MQTT订阅失败: {}", 
                            config != null ? config.getId() : "unknown", e.getMessage(), e);
                }
            }
            
            log.info("默认MQTT订阅注册完成，成功: {}/总计: {}", successCount, configs.size());
        } catch (Exception e) {
            log.error("注册默认MQTT订阅失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 添加新的订阅主题
     * 
     * @param configId MQTT配置ID
     * @param topic 主题
     * @param qos QoS级别
     * @return 是否成功
     */
    public boolean addSubscription(Long configId, String topic, int qos) {
        try {
            MqttPahoMessageDrivenChannelAdapter adapter = inboundAdapterMap.get(configId);
            if (adapter != null) {
                adapter.addTopic(topic, qos);
                log.info("成功添加订阅主题: {}, configId: {}", topic, configId);
                return true;
            } else {
                log.error("未找到MQTT适配器，无法添加订阅, configId: {}", configId);
                return false;
            }
        } catch (Exception e) {
            log.error("添加订阅主题失败: {}, configId: {}", topic, configId, e);
            return false;
        }
    }
    
    /**
     * 移除订阅主题
     * 
     * @param configId MQTT配置ID
     * @param topic 主题
     * @return 是否成功
     */
    public boolean removeSubscription(Long configId, String topic) {
        try {
            MqttPahoMessageDrivenChannelAdapter adapter = inboundAdapterMap.get(configId);
            if (adapter != null) {
                adapter.removeTopic(topic);
                log.info("成功移除订阅主题: {}, configId: {}", topic, configId);
                return true;
            } else {
                log.error("未找到MQTT适配器，无法移除订阅, configId: {}", configId);
                return false;
            }
        } catch (Exception e) {
            log.error("移除订阅主题失败: {}, configId: {}", topic, configId, e);
            return false;
        }
    }
} 